package com.yao.dubboconsumer.discovery.impl;

import com.yao.dubboapi.constant.ZookeeperConstant;
import com.yao.dubboconsumer.discovery.ServiceDiscovery;
import com.yao.dubboconsumer.loadbalance.LoadBalance;
import com.yao.dubboconsumer.loadbalance.impl.RandomLoadBalance;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.retry.ExponentialBackoffRetry;

import java.util.ArrayList;
import java.util.List;

/**
 * @author Henry Yao
 * @since 2021/5/3
 */
public class ServiceDiscoveryImpl implements ServiceDiscovery {
    private final List<LoadBalance> loadBalanceList = new ArrayList<>();
    private final CuratorFramework curatorFramework;
    private List<String> servers;

    {
        loadBalanceList.add(new RandomLoadBalance());
        // todo  增加其他负载均衡方式
    }

    public ServiceDiscoveryImpl() {
        curatorFramework = CuratorFrameworkFactory.builder()
                .namespace(ZookeeperConstant.ZK_ROOT)
                .connectString(ZookeeperConstant.ZK_ADDRESS)
                .connectionTimeoutMs(5000)
                .sessionTimeoutMs(5000)
                .retryPolicy(new ExponentialBackoffRetry(2000, 3))
                .build();
        curatorFramework.start();
    }

    @Override
    public String discovery(String serviceName, LoadBalance.Type type) throws Exception {
        // 缓存不存在则去zk获取并放入缓存
        // 第一次获取
        if (servers == null) {
            String serviceNode = "/" + serviceName;
            if (curatorFramework.checkExists().forPath(serviceNode) == null) {
                System.out.println("未找到服务：" + serviceName);
                return null;
            }
            servers = curatorFramework.getChildren().forPath(serviceNode);
            // 注册监听及时更新提供者列表
            addWatcher(serviceNode);
        }
        if (servers.size() == 0) {
            System.out.println("未找到提供者：" + serviceName);
            return null;
        }
        return balance(servers, type);

    }

    /**
     * 监听子节点变化
     * @param serviceNode
     * @throws Exception
     */
    private void addWatcher(String serviceNode) throws Exception {
        PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFramework, serviceNode, true);
        pathChildrenCache.getListenable().addListener((curatorFramework, pathChildrenCacheEvent) -> {
            servers = curatorFramework.getChildren().forPath(serviceNode);
        });
        pathChildrenCache.start();
    }

    /**
     * 进行负载均衡
     *
     * @param servers
     * @param type
     * @return
     */
    private String balance(List<String> servers, LoadBalance.Type type) {
        String server;
        for (LoadBalance lb : loadBalanceList) {
            server = lb.doBalance(servers, type);
            if (server != null) {
                return server;
            }
        }
        return null;
    }
}
